package query;

import java.io.IOException;
import java.util.ArrayList;

import libMapReduce.MultiTableInputCollection;
import libMapReduce.TableMapReduceUtil;



import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
//import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;

public class MapJoin {

    static class Mapper1 extends TableMapper<ImmutableBytesWritable, IntWritable> {

        private int numRecords = 0;
        private static final IntWritable one = new IntWritable(1);

        @Override
        public void map(ImmutableBytesWritable row, Result values, Context context) throws IOException {
            // extract userKey from the compositeKey (userId + counter)
//            ImmutableBytesWritable userKey = new ImmutableBytesWritable(row.get(), 0, Bytes.SIZEOF_INT);
            ImmutableBytesWritable userKey = new ImmutableBytesWritable(row.get(), 0, 1);
                        	System.out.println("key:"+row.toString());
			//            	System.out.println();
			            	KeyValue v = values.getColumnLatest(Bytes.toBytes("cf1"), Bytes.toBytes("c1"));
			                System.out.println(Bytes.toString(v.getValue()));
			//            	context.write(new ImmutableBytesWritable(v.getValue()), one);
//            System.out.println(userKey);
            numRecords++;
            if ((numRecords % 10000) == 0) {
            	
                context.setStatus("mapper processed " + numRecords + " records so far");
                System.out.println(System.currentTimeMillis());
            }
        }
    }

    public static class Reducer1 extends TableReducer<ImmutableBytesWritable, IntWritable, ImmutableBytesWritable> {

        public void reduce(ImmutableBytesWritable key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            return;
            /*
            for (IntWritable val : values) {
                sum += val.get();
            }

            Put put = new Put(key.get());
            put.add(Bytes.toBytes("cf1"), Bytes.toBytes("total"), Bytes.toBytes(sum));
//            System.out.println(String.format("stats :   key : %s  count : %d", key.get().toString(), sum));
            context.write(key, put);
            */
        }
    }
    
    public static void main(String[] args) throws Exception {
    	Configuration configuration = HBaseConfiguration.create();//new HBaseConfiguration();
        
        

//        String columns = "details"; // comma seperated
//        scan.addColumns(columns);
//        scan.addColumns(columns);
//        configuration = HBaseConfiguration.create();
		configuration.set("hbase.master", "192.9.200.169:44696");
		configuration.set("hbase.zookeeper.quorum","192.9.200.169");  
		configuration.set("hbase.zookeeper.property.clientPort", "2181");    
		configuration.set("hbase.HServerAddress", "192.9.200.169:44696");
		Job job = new Job(configuration, "Hbase_FreqCounter1");
        job.setJarByClass(MapJoin.class);
		
//        scan.setFilter(new FirstKeyOnlyFilter());
        String table = "DH_10K";//args[0];//input the table dynamic table name in argument
		System.out.println("Testing on:" + table);
		Scan scan1 = new Scan();
		scan1.setStartRow(null);
		scan1.setStopRow(null);

		
		String table2 = "DH_1MR";
		Scan scan2 = new Scan();
		scan2.setStartRow(null);
		scan2.setStopRow(null);

		
		ArrayList<String> columnFamilies = new ArrayList<String>();
		columnFamilies.add("cf1");
		columnFamilies.add("cf2");
		columnFamilies.add("cf3");
		columnFamilies.add("cf4");
		columnFamilies.add("cf5");
		columnFamilies.add("cf6");
		columnFamilies.add("cf7");
		columnFamilies.add("cf8");
		columnFamilies.add("cf9");
		columnFamilies.add("cf10");
		
		ArrayList<String> clms = new ArrayList<String>();
		clms.add("c1");
		clms.add("c2");
		clms.add("c3");
		clms.add("c4");
		clms.add("c5");
		clms.add("c6");
		clms.add("c7");
		clms.add("c8");
		clms.add("c9");
		clms.add("c10");
		
		
		scan1.addColumn(columnFamilies.get(0).getBytes(), clms.get(0).getBytes());
		scan1.addColumn(columnFamilies.get(1).getBytes(), clms.get(1).getBytes());
//		scan1.addColumn(columnFamilies.get(2).getBytes(), clms.get(2).getBytes());
//		scan1.addColumn(columnFamilies.get(3).getBytes(), clms.get(3).getBytes());
//		scan1.addColumn(columnFamilies.get(4).getBytes(), clms.get(4).getBytes());
//		scan1.addColumn(columnFamilies.get(5).getBytes(), clms.get(5).getBytes());
//		scan1.addColumn(columnFamilies.get(6).getBytes(), clms.get(6).getBytes());
//		scan1.addColumn(columnFamilies.get(7).getBytes(), clms.get(7).getBytes());
//		scan1.addColumn(columnFamilies.get(8).getBytes(), clms.get(8).getBytes());
//		scan1.addColumn(columnFamilies.get(9).getBytes(), clms.get(9).getBytes());

//		scan1.addColumn(Bytes.toBytes("cf3"), Bytes.toBytes("c3"));
		
//		scan2.addFamily(cf1.getBytes());
//		scan2.addFamily(cf2.getBytes());
//		scan2.addFamily(cf3.getBytes());
//		scan2.addColumn(cf1.getBytes(), c1.getBytes());
//		scan2.addColumn(cf2.getBytes(), c2.getBytes());
//		scan1.addFamily(family22.getBytes());
//		scan2.addColumn(family2.getBytes(), column2.getBytes());
		MultiTableInputCollection mit = new MultiTableInputCollection();
		
		scan1.setCacheBlocks(false);
		scan2.setCacheBlocks(false);
		mit.Add(table, scan1);
//		mit.Add(table2, scan2);
		
		
		
//		TableMapReduceUtil.initTableMapperJob(table1.getBytes(), scan1, Mapper1.class, ImmutableBytesWritable.class, IntWritable.class, job);
//		TableMapReduceUtil.initTableMapperJob(table1.getBytes(), scan2, Mapper1.class, ImmutableBytesWritable.class, IntWritable.class, job);
		
		TableMapReduceUtil.initTableMapperJob(mit, Mapper1.class, ImmutableBytesWritable.class,
				IntWritable.class, job);
		
//        TableMapReduceUtil.initTableMapperJob("test", scan, Mapper1.class, ImmutableBytesWritable.class,
//                IntWritable.class, job);
//        TableMapReduceUtil.initTableMapperJob("mytable", scan, Mapper1.class, ImmutableBytesWritable.class,
//                IntWritable.class, job);

        TableMapReduceUtil.initTableReducerJob("result", Reducer1.class, job);
        long startTime = System.currentTimeMillis();
        System.out.println(startTime);
        job.waitForCompletion(true);
        System.out.println("Running time:"+(System.currentTimeMillis()-startTime)/1000.00 + "seconds");
    }
}